package com.xtxb.app.rmq.consume;

import com.xtxb.app.rmq.common.MyMQConsume;
import com.xtxb.app.rmq.common.Topic;
import com.xtxb.app.rmq.conf.MQServer;
import com.xtxb.app.rmq.conf.MQTopic;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

@Component
public class ConsumeServer {
    private static Logger log= LoggerFactory.getLogger(ConsumeServer.class);

    /**
     * {
     *     serverName_groupName_topic:{
     *         tag:uuid
     *     }
     * }
     */
    private Map<String,Map<String,String>> consumerTopics=new ConcurrentHashMap<>();
    private Map<String, MyMQConsume> consumerMap=new HashMap<>();
    @Autowired
    private MQServer server;
    @Autowired
    private MQTopic topicServer;
    @Autowired
    private WebSocktSender sender;
    private SimpleDateFormat sdf=new SimpleDateFormat("yyyyMMddHHmmss");

    public synchronized MyMQConsume getConsume(String groupName,String server){
        MyMQConsume consumer=consumerMap.get(server + "_" + groupName);
        if(consumer==null) {
            consumer = new MyMQConsume(groupName);
            consumer.setNamesrvAddr(server);
//        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP);
            consumer.setConsumeTimestamp(sdf.format(new Date()));
            consumer.setMessageModel(MessageModel.BROADCASTING);
            try {
                consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
                    try {
                        for (Message msg : msgs) {
                            //消费者获取消息 这里只输出 不做后面逻辑处理
                            String body = new String(msg.getBody(), "utf-8");
                            log.info("接收消息: " + msg.getTopic() + "\t" + msg.getProperty("TAGS") + "\t" + body);
                            Map<String,String> tags=consumerTopics.get(server+"_"+groupName+"_"+msg.getTopic());

                            if(tags!=null) {
                                for(String tag: tags.keySet()) {
                                    if(tag.equals("*") || tag.equals(msg.getProperty("TAGS"))) {
                                        sender.send(tags.get(tag), body);
                                    }
                                }
                            }
                        }
                    } catch (UnsupportedEncodingException e) {
                        e.printStackTrace();
                    }
                    return
                            ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                });
                consumer.start();
                consumerMap.put(server + "_" + groupName, consumer);
            } catch (MQClientException e) {
                e.printStackTrace();
            }
        }
        return consumer;
    }


    public boolean startConsume(String uuid){
        Topic topic=topicServer.getTopic(uuid);
        /**不存在主题*/
        if(topic==null) {
            log.error("不存在UUID:{} 的主题",uuid);
            return false;
        }
        String serverAddr=server.getServer(topic.getServerName());
        MyMQConsume consumer=getConsume(topic.getGroupName(),serverAddr);
        /**获取Consume失败*/
        if(consumer==null){
            log.error("创建Consume失败:{} {}",serverAddr,topic.getGroupName());
            return false;
        }
        String tag=StringUtils.isEmpty(topic.getTag())?"*":topic.getTag();
        Map<String,String> tags=consumerTopics.get(serverAddr+"_"+topic.getGroupName()+"_"+topic.getTopic());
        /**还未订阅主题*/
        if(tags==null ) {
            tags=new ConcurrentHashMap<>();
            try {
                consumer.subscribe(topic.getTopic(),"*");
                tags.put(tag,uuid);
                consumerTopics.put(serverAddr+"_"+topic.getGroupName()+"_"+topic.getTopic(),tags);
            } catch (MQClientException e) {
                log.error("订阅消息失败",e);
                return false;
            }
        }else if(tags.containsKey(tag)){
            /**已经订阅*/
            log.error("Topic已经订阅:{} {} {}",serverAddr,topic.getGroupName(),topic.getTopic());
            return false;
        }else{
            tags.put(tag,uuid);
        }
        return true;
    }

    public boolean stopConsume(String uuid){
        Topic topic=topicServer.getTopic(uuid);
        if(topic==null) {
            log.error("不存在UUID:{} 的主题",uuid);
            return false;
        }
        String serverAddr=server.getServer(topic.getServerName());
        String tag=StringUtils.isEmpty(topic.getTag())?"*":topic.getTag();
        Map<String,String> tags=consumerTopics.get(serverAddr+"_"+topic.getGroupName()+"_"+topic.getTopic());
        if(tags!=null && tags.containsKey(tag)){
            if(tags.size()==1){
                getConsume(topic.getGroupName(),serverAddr).unsubscribe(topic.getTopic());
                consumerTopics.remove(serverAddr+"_"+topic.getGroupName()+"_"+topic.getTopic());
                for(String key : consumerMap.keySet().toArray(new String[0])){
                    if(consumerTopics.keySet().stream().anyMatch(string->string.startsWith(key))) {
                        consumerMap.remove(key).shutdown();
                    }
                }
            }else{
                tags.remove(tag);
            }
            return true;
        }else{
            return false;
        }
    }


    public Map<String,Map<String,String>> getCconsumerTopics(){
        return consumerTopics;
    }
}
